package app;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import utils.MykafkaUtil;

import java.text.SimpleDateFormat;

public class BaseLogApp {


    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(4);
        env.enableCheckpointing(5*1000L);
        env.getCheckpointConfig().setCheckpointTimeout(60*1000L);
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,3000L));
        System.setProperty("HADOOP_USER_NAME","atguigu");
        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/ck3",false));
        String topic ="ods_base_log";
        String groupId ="base_log_app_groupId";
        FlinkKafkaConsumer<String> kafkaSource = MykafkaUtil.getKafkaSource(topic, groupId);
        DataStreamSource<String> streamSource = env.addSource(kafkaSource);
        SingleOutputStreamOperator<JSONObject> JsonDS = streamSource.map(line -> JSON.parseObject(line));
        KeyedStream<JSONObject, String> keyedStream = JsonDS.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject jsonObject) throws Exception {
                return jsonObject.getJSONObject("common").getString("mid");
            }
        });
        SingleOutputStreamOperator<JSONObject> streamOperator = keyedStream.map(new RichMapFunction<JSONObject, JSONObject>() {
            private ValueState<String> lastVisitDateState;
            private SimpleDateFormat sdf;

            @Override
            public void open(Configuration parameters) throws Exception {

                super.open(parameters);
                sdf = new SimpleDateFormat("YYYY-MM-dd");
                lastVisitDateState = getRuntimeContext().getState(new ValueStateDescriptor<String>("state", String.class));

            }

            @Override
            public JSONObject map(JSONObject value) throws Exception {
                String isNew = value.getJSONObject("common").getString("is_new");
                // 获取当前访问日期
                String curr = sdf.format(value.getLong("ts"));
                if (isNew.equals("1")) {
                    String lastVisitDate = lastVisitDateState.value();

                    if (lastVisitDate != null && lastVisitDate.equals(curr)) {
                        isNew = "0";
                        value.getJSONObject("common").put("is_new", isNew);
                    }
                } else {
                    lastVisitDateState.update(curr);
                }
                return value;
            }
        });
        streamOperator.print();


        env.execute();
    }

}
